package glue_datasyncer

import (
	"encoding/json"
	"fmt"
	"github.com/gogo/protobuf/proto"
	"github.com/golang/snappy"
	"github.com/prometheus/prometheus/prompb"
	streamCtx "gluenet/internal/glue-datasyncer/context"
	"sort"
	"strings"
	"sync"
)

func MetricsRouter(c *streamCtx.StreamContext) {
	for {
		select {
		case <-c.Context.Done():
			return
		case data := <-c.MetricsChan:
			reqBuf, err := snappy.Decode(nil, data)
			if err != nil {
				c.Logger.Errorf("snappy decode failed %v", err)
				return
			}
			var req prompb.WriteRequest
			if err = proto.Unmarshal(reqBuf, &req); err != nil {
				c.Logger.Errorf("proto unmarshal failed %v", err)
				return
			}

			//c.Logger.Debugf("req %v", req)

			for _, ts := range req.Timeseries {
				var kv = make([]string, len(ts.Labels))
				var ms = make(map[string]string, len(ts.Labels))
				for k, label := range ts.Labels {
					kv[k] = strings.Join([]string{label.Name, label.Value}, "=")
					ms[label.Name] = label.Value
				}
				sort.Strings(kv)
				keys := strings.Join(kv, "|")
				c.MetricLabels.Store(keys, ms)
				chsi, _ := c.MetricSubChan.LoadOrStore(keys, &sync.Map{})

				var sd = make(map[string]interface{}, len(ms)+2)
				for name, value := range ms {
					sd[name] = value
				}
				sd["value"] = ts.Samples[0].Value
				sd["time"] = ts.Samples[0].Timestamp / 1000
				bts, _ := json.Marshal(sd)
				chs, _ := chsi.(*sync.Map)
				chs.Range(func(key, value interface{}) bool {
					if (func(ch chan string) bool {
						select {
						case _, received := <-ch:
							return !received
						default:
						}
						return false
					})(value.(chan string)) {
						return false
					} else {
						value.(chan string) <- string(bts)
						return true
					}
				})
			}
		}
	}
}

func LogsRouter(c *streamCtx.StreamContext) {
	for {
		select {
		case <-c.Context.Done():
			return
		case data := <-c.LogsChan:
			fmt.Println("recive log", data)
		}
	}
}
